package com.songyang.yougnsql.core.data.recover;

import com.songyang.yougnsql.common.ExceptionHandler;
import com.songyang.yougnsql.core.data.dataItem.DataItem;
import com.songyang.yougnsql.core.data.logger.Logger;
import com.songyang.yougnsql.core.data.page.Page;
import com.songyang.yougnsql.core.data.page.PageOne;
import com.songyang.yougnsql.core.data.page.PageX;
import com.songyang.yougnsql.core.data.pagecache.PageCache;
import com.songyang.yougnsql.core.utils.Address;
import com.songyang.yougnsql.core.utils.ByteUtils;
import com.songyang.yougnsql.core.transaction.TransactionManager;

import java.util.*;

/**
 * @ClassName Recover
 * @Description
 * @date 2022/12/12 11:17
 * @Author yanceysong
 * @Version 1.0
 */
public class Recover {
    private static final byte LOG_TYPE_INSERT = 0;
    private static final byte LOG_TYPE_UPDATE = 1;
    private static final int REDO = 0;
    private static final int UNDO = 1;
    /**
     * [LogType] [XID] [UID] [OldRaw] [NewRaw]
     */
    private static final int OF_TYPE = 0;
    private static final int OF_XID = OF_TYPE + 1;
    private static final int OF_UPDATE_UID = OF_XID + 8;
    private static final int OF_UPDATE_RAW = OF_UPDATE_UID + 8;
    /**
     * [LogType] [XID] [page_number] [Offset] [Raw]
     */
    private static final int OF_INSERT_PAGE_NUMBER = OF_XID + 8;
    private static final int OF_INSERT_OFFSET = OF_INSERT_PAGE_NUMBER + 4;
    private static final int OF_INSERT_RAW = OF_INSERT_OFFSET + 2;

    /**
     * 数据库启动时从日志恢复数据，
     *
     * @param tm        事务管理
     * @param logger    日志
     * @param pageCache 页面缓存
     */
    public static void recover(TransactionManager tm, Logger logger, PageCache pageCache) {
        //日志的指针移向开头
        logger.rewind();
        int maxPageNumber = PageOne.PAGE_ONE_NUMBER;
        while (true) {
            byte[] log = logger.nextLog();
            if (log == null) break;
            int pageNumber = 0;
            if (isInsertLog(log)) {
                InsertLogInfo insertLogInfo = parseInsertLog(log);
                pageNumber = insertLogInfo.getPageNumber();
            } else if (isUpdateLog(log)) {
                UpdateLogInfo updateLogInfo = parseUpdateLog(log);
                pageNumber = updateLogInfo.getPageNumber();
            }
            maxPageNumber = Math.max(maxPageNumber, pageNumber);
        }
        //排除错误的页面数据
        pageCache.truncateByPageNumber(maxPageNumber);
        //恢复已经提交事务
        redoTransaction(tm, logger, pageCache);
        //撤销未完成的事务
        undoTransaction(tm, logger, pageCache);
    }

    /**
     * 重做已经完成的事务
     *
     * @param tm        事务管理
     * @param logger    日志
     * @param pageCache 页面缓存
     */
    private static void redoTransaction(TransactionManager tm, Logger logger, PageCache pageCache) {
        logger.rewind();
        while (true) {
            byte[] log = logger.nextLog();
            if (log == null) break;
            if (isInsertLog(log)) {
                InsertLogInfo insertLogInfo = parseInsertLog(log);
                //如果这个事务已经提交了
                if (tm.isActive(insertLogInfo.getXid())) {
                    doInsertLog(pageCache, log, REDO);
                }
            } else if (isUpdateLog(log)) {
                UpdateLogInfo updateLogInfo = parseUpdateLog(log);
                if (tm.isActive(updateLogInfo.getXid())) {
                    doUpdateLog(pageCache, log, REDO);
                }
            }
        }
    }

    /**
     * 恢复更新日志的数据
     *
     * @param pageCache 页面缓存
     * @param log       日志数据
     * @param flag      标志位
     */
    private static void doUpdateLog(PageCache pageCache, byte[] log, int flag) {
        int pageNumber;
        short offset;
        byte[] raw;
        if (flag == REDO) {
            UpdateLogInfo updateLogInfo = parseUpdateLog(log);
            pageNumber = updateLogInfo.getPageNumber();
            offset = updateLogInfo.getOffset();
            raw = updateLogInfo.getNewRaw();
        } else {
            //如果是update日志的UNDO，那么需要将页面中的数据修改为旧数据
            UpdateLogInfo updateLogInfo = parseUpdateLog(log);
            pageNumber = updateLogInfo.getPageNumber();
            offset = updateLogInfo.getOffset();
            raw = updateLogInfo.getOldRaw();
        }
        Page page = null;
        try {
            page = pageCache.getPage(pageNumber);
            //更新
            PageX.update(page, raw, offset);
        } catch (Exception e) {
            ExceptionHandler.handler(e);
        } finally {
            assert page != null;
            //释放页面引用
            page.release();
        }
    }

    /**
     * 恢复插入日志的数据
     *
     * @param pageCache 页面缓存
     * @param log       日志数据
     * @param flag      标志位
     */
    private static void doInsertLog(PageCache pageCache, byte[] log, int flag) {
        InsertLogInfo insertLogInfo = parseInsertLog(log);
        Page page = null;
        try {
            page = pageCache.getPage(insertLogInfo.getPageNumber());
            // 如果是insert日志的UNDO，那么只需要将DataItem的是否有效标志位设置为无效即可
            if (flag == UNDO) {
                DataItem.setDataItemRawInvalid(insertLogInfo.getRaw());
            }
            //开始插入数据
            PageX.insert(page, insertLogInfo.getRaw(), insertLogInfo.getOffset());
        } catch (Exception e) {
            ExceptionHandler.handler(e);
        } finally {
            //从缓存当中取消这个引用
            assert page != null;
            page.release();
        }
    }

    /**
     * 撤销未完成的事务
     *
     * @param tm        事务管理
     * @param logger    日志
     * @param pageCache 页面缓存
     */
    private static void undoTransaction(TransactionManager tm, Logger logger, PageCache pageCache) {
        Map<Long, List<byte[]>> logCache = new HashMap<>();
        logger.rewind();
        while (true) {
            byte[] log = logger.nextLog();
            if (log == null) break;
            if (isInsertLog(log)) {
                InsertLogInfo li = parseInsertLog(log);
                long xid = li.getXid();
                if (tm.isActive(xid)) {
                    if (!logCache.containsKey(xid)) {
                        logCache.put(xid, new ArrayList<>());
                    }
                    logCache.get(xid).add(log);
                }
            } else {
                UpdateLogInfo xi = parseUpdateLog(log);
                long xid = xi.getXid();
                if (tm.isActive(xid)) {
                    if (!logCache.containsKey(xid)) {
                        logCache.put(xid, new ArrayList<>());
                    }
                    logCache.get(xid).add(log);
                }
            }
        }
        // 对所有active log进行倒序undo
        for (Map.Entry<Long, List<byte[]>> entry : logCache.entrySet()) {
            List<byte[]> logs = entry.getValue();
            for (int i = logs.size() - 1; i >= 0; i--) {
                byte[] log = logs.get(i);
                if (isInsertLog(log)) {
                    doInsertLog(pageCache, log, UNDO);
                } else {
                    doUpdateLog(pageCache, log, UNDO);
                }
            }
            tm.abort(entry.getKey());
        }
    }

    /**
     * 判断是否是插入的log
     *
     * @param log log
     * @return 结果
     */
    private static Boolean isInsertLog(byte[] log) {
        return log[OF_TYPE] == LOG_TYPE_INSERT;
    }

    /**
     * 判断是否是更新的log
     *
     * @param log log
     * @return 结果
     */
    private static Boolean isUpdateLog(byte[] log) {
        return log[OF_TYPE] == LOG_TYPE_UPDATE;
    }

    /**
     * 如果是插入的日志，则解析为插入日志的info
     *
     * @param log log数据
     * @return 插入日志的info
     */
    private static InsertLogInfo parseInsertLog(byte[] log) {
        InsertLogInfo insertLogInfo = new InsertLogInfo();
        insertLogInfo.setXid(ByteUtils.parseLong(Arrays.copyOfRange(log, OF_XID, OF_INSERT_PAGE_NUMBER)));
        insertLogInfo.setPageNumber(ByteUtils.parseInt(Arrays.copyOfRange(log, OF_INSERT_PAGE_NUMBER, OF_INSERT_OFFSET)));
        insertLogInfo.setOffset(ByteUtils.parseShort(Arrays.copyOfRange(log, OF_INSERT_OFFSET, OF_INSERT_RAW)));
        insertLogInfo.setRaw(Arrays.copyOfRange(log, OF_INSERT_RAW, log.length));
        return insertLogInfo;
    }

    /**
     * 如果是插入的日志，则解析为插入更新日志的info
     *
     * @param log log数据
     * @return 更新日志的info
     */
    private static UpdateLogInfo parseUpdateLog(byte[] log) {
        UpdateLogInfo updateLogInfo = new UpdateLogInfo();
        updateLogInfo.setXid(ByteUtils.parseLong(Arrays.copyOfRange(log, OF_XID, OF_UPDATE_UID)));
        long uid = ByteUtils.parseLong(Arrays.copyOfRange(log, OF_UPDATE_UID, OF_UPDATE_RAW));
        updateLogInfo.setOffset(Address.getOffsetByUid(uid));
        updateLogInfo.setPageNumber(Address.getPageNumberByUid(uid));
        int length = (log.length - OF_UPDATE_RAW) / 2;
        updateLogInfo.setOldRaw(Arrays.copyOfRange(log, OF_UPDATE_RAW, OF_UPDATE_RAW + length));
        updateLogInfo.setNewRaw(Arrays.copyOfRange(log, OF_UPDATE_RAW + length, OF_UPDATE_RAW + length * 2));
        return updateLogInfo;
    }

    /**
     * 组装一条update的日志
     * [LogType] [XID] [UID] [OldRaw] [NewRaw]
     *
     * @param xid      事务的xid
     * @param dataItem dataItem
     * @return 一条update日志数组
     */
    public static byte[] updateLog(long xid, DataItem dataItem) {
        byte[] logType = {LOG_TYPE_UPDATE};
        byte[] xidRaw = ByteUtils.long2Byte(xid);
        byte[] uidRaw = ByteUtils.long2Byte(dataItem.getUid());
        byte[] oldRaw = dataItem.getOldRaw();
//        byte[] raw = dataItem.getRaw();
//        byte[] newRaw = Arrays.copyOfRange(raw, 0, raw.length);
        byte[] newRaw = dataItem.getRaw();
        return ByteUtils.concat(logType, xidRaw, uidRaw, oldRaw, newRaw);
    }

    /**
     * 组装一条 插入日志
     * [LogType] [XID] [page_number] [Offset] [Raw]
     *
     * @param xid  事务的id
     * @param page 页面
     * @param raw  插入的数据
     * @return 一条insert的日志数组
     */
    public static byte[] getInsertLogBytes(long xid, Page page, byte[] raw) {
        byte[] logTypeRaw = {LOG_TYPE_INSERT};
        byte[] xidRaw = ByteUtils.long2Byte(xid);
        byte[] pageNumberRaw = ByteUtils.int2Byte(page.getPageNumber());
        byte[] offsetRaw = ByteUtils.short2Byte(PageX.getOffset(page));
        return ByteUtils.concat(logTypeRaw, xidRaw, pageNumberRaw, offsetRaw, raw);
    }
}
